from pyspark.sql import SparkSession
from pyspark.sql.functions import countDistinct, sum, count, col, when
from pyspark.sql.types import IntegerType

# 创建 SparkSession
spark = SparkSession.builder \
    .appName("UserLogAnalysis") \
    .getOrCreate()

# 从本地文件系统读取 CSV 文件
df = spark.read.csv("file:///home/jkx/small_user_log.csv", header=True, inferSchema=True)

# 确保关键列的数据类型为整数类型
df = df.withColumn("action", col("action").cast(IntegerType()))
df = df.withColumn("brand_id", col("brand_id").cast(IntegerType()))
df = df.withColumn("gender", col("gender").cast(IntegerType()))

print("数据基本信息：")
df.printSchema()

# 查看数据集行数
rows = df.count()
print("表内的行数据数量为", rows)

# 查询双 11 那天有多少人购买了商品（去掉同一个 user_id 的重复购买）
unique_buyers = df.filter((col("month") == 11) & (col("day") == 11) & (col("action") == 2)) \
    .agg(countDistinct("user_id").alias("unique_buyers")) \
    .collect()[0]["unique_buyers"]
print("双 11 那天购买商品的不重复用户数量为", unique_buyers)

# 给定品牌 brand_id 为 2661，求双 11 当天此品牌商品的销售数量
# 假设每条记录代表一次购买数量为 1
brand_sales = df.filter((col("month") == 11) & (col("day") == 11) & (col("action") == 2) & (col("brand_id") == 2661)) \
    .agg(count("*").alias("brand_sales")) \
    .collect()[0]["brand_sales"]
print("双 11 当天品牌 2661 的商品销售数量为", brand_sales)

# 查询双 11 那天女性购买商品的数量
# 假设每条记录代表一次购买数量为 1
female_purchases = df.filter((col("month") == 11) & (col("day") == 11) & (col("action") == 2) & (col("gender") == 0)) \
    .agg(count("*").alias("female_purchases")) \
    .collect()[0]["female_purchases"]
print("双 11 那天女性购买商品的数量为", female_purchases)

# 根据收货地址，统计各个省双 11 那天用户下单的总次数，并进行排序
province_orders = df.filter((col("month") == 11) & (col("day") == 11) & (col("action") == 1)) \
    .groupBy("province") \
    .agg(count("*").alias("order_count")) \
    .orderBy(col("order_count").desc())
print("双 11 那天各个省用户下单总次数统计结果：")
province_orders.show()

# 停止 SparkSession
spark.stop()